--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 0d540279c0c2180d98f8d741894d2f68c807477f
Parents : 0d8cbd2
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-13T17:34:49-05:00
feat(async_utils): update coroutine scheduling with thread safety and future management
Changes
Diff
diff --git a/meshchatx/src/backend/async_utils.py b/meshchatx/src/backend/async_utils.py
index e513571f..016a293b 100644
--- a/meshchatx/src/backend/async_utils.py
+++ b/meshchatx/src/backend/async_utils.py
@@ -1,11 +1,14 @@
import asyncio
import sys
+import threading
from collections.abc import Coroutine
class AsyncUtils:
- # remember main loop
main_loop: asyncio.AbstractEventLoop | None = None
+ _pending_futures: list = []
+ _futures_lock = threading.Lock()
+ _FUTURES_SWEEP_THRESHOLD = 64
@staticmethod
def apply_asyncio_313_patch():
@@ -16,7 +19,6 @@ class AsyncUtils:
if sys.version_info >= (3, 13):
import asyncio.base_events
- # Patch sendfile so aiohttp uses its SSL fallback on 3.13.
original_sendfile = asyncio.base_events.BaseEventLoop.sendfile
async def patched_sendfile(
@@ -47,14 +49,23 @@ class AsyncUtils:
def set_main_loop(loop: asyncio.AbstractEventLoop):
AsyncUtils.main_loop = loop
- # this method allows running the provided async coroutine from within a sync function
- # it will run the async function on the main event loop if possible, otherwise it logs a warning
@staticmethod
def run_async(coroutine: Coroutine):
- # run provided coroutine on main event loop, ensuring thread safety
+ """Schedule *coroutine* on the main event loop from any thread.
+
+ Returned futures are tracked so they (and the closures they reference)
+ can be garbage-collected promptly once finished.
+ """
if AsyncUtils.main_loop and AsyncUtils.main_loop.is_running():
- asyncio.run_coroutine_threadsafe(coroutine, AsyncUtils.main_loop)
+ future = asyncio.run_coroutine_threadsafe(
+ coroutine, AsyncUtils.main_loop,
+ )
+ with AsyncUtils._futures_lock:
+ AsyncUtils._pending_futures.append(future)
+ if len(AsyncUtils._pending_futures) >= AsyncUtils._FUTURES_SWEEP_THRESHOLD:
+ AsyncUtils._pending_futures = [
+ f for f in AsyncUtils._pending_futures if not f.done()
+ ]
return
- # main event loop not running...
print("WARNING: Main event loop not available. Could not schedule task.")
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────